package com.hxq.demo1;

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.TopicPartition;

import java.time.Duration;
import java.util.*;

/**自动提交：消费者可能会丢数据，如：消费到100条数据，自动提交时间是1秒，offset被提交了，但是还没有消费完数据，此时消费者挂掉了。当消费者再次启动，会从101继续消费
 * 手动提交：
 * 虽然自动提交 offset 十分简介便利，但由于其是基于时间提交的，开发人员难以把握
 * offset 提交的时机。因此 Kafka 还提供了手动提交 offset 的 API。
 * 手动提交 offset 的方法有两种：分别是 commitSync（同步提交）和 commitAsync（异步
 * 提交）。两者的相同点是，都会将本次 poll 的一批数据最高的偏移量提交；不同点是，
 * commitSync 阻塞当前线程，一直到提交成功，并且会自动失败重试（由不可控因素导致，
 * 也会出现提交失败）；而 commitAsync 则没有失败重试机制，故有可能提交失败。
 * @author hxq
 * @date 2022/4/14 18:08
 */
public class CustomConsumer {

    public static void main(String[] args) {
        autoCommitOffset();
//        manualCommitSync();
//        manualCommitAsync();
    }

    private static void autoCommitOffset(){

        Properties properties = setConfigInfo(true);
//        Properties properties = setConfigInfo(false);//如果不开启自动提交，也不设置手动提交，则不会提交offset,所以每次启动消费者都会消费到上一个offset的数据
        //创建消费者
        KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(properties);
        //订阅主题（指定分区消费）
        List<TopicPartition> topicPartitions = new ArrayList<>();
        topicPartitions.add(new TopicPartition("first",0));
        consumer.assign(topicPartitions);

        while (true){
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
            for (ConsumerRecord<String, String> record : records){
                System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
            }

        }

    }

    //同步手动提交（会自动失败重试）
    private static void manualCommitSync(){
        Properties properties = setConfigInfo(false);
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(properties);
        consumer.subscribe(Arrays.asList("first"));//消费者订阅主题
        //消费数据

        while (true) {
            //消费者拉取数据
            ConsumerRecords<String, String> records =
                    consumer.poll(Duration.ofMillis(100));
            for (ConsumerRecord<String, String> record : records) {
                System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
            }
            //同步提交，当前线程会阻塞直到 offset 提交成功
            consumer.commitSync();

        }
    }

    //异步手动提交
    private static void manualCommitAsync(){
        Properties properties = setConfigInfo(false);
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(properties);
        consumer.subscribe(Arrays.asList("first"));//消费者订阅主题

        while (true){
            ConsumerRecords<String, String> records =
                    consumer.poll(Duration.ofMillis(100));//消费者拉取数据
            for (ConsumerRecord<String, String> record : records) {
                System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
            }

            //异步提交
            consumer.commitAsync(new OffsetCommitCallback() {
                @Override
                public void onComplete(Map<TopicPartition, OffsetAndMetadata> map, Exception e) {
                    if (e!=null){
                        System.err.println("Commit failed for" +map);
                    }
                }
            });
        }

    }




    private static Properties setConfigInfo(Boolean isAuto){
        Properties props = new Properties();
        props.put("bootstrap.servers", "212.129.155.35:9092");
        props.put("group.id", "test");
        props.put("enable.auto.commit", isAuto);//是否开启自动提交 offset 功能
        props.put("auto.commit.interval.ms", "1000");//自动提交 offset 的时间间隔
        props.put("key.deserializer",
                "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer",
                "org.apache.kafka.common.serialization.StringDeserializer");

//        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,"earliest");//从头消费 相当于控制台上 加 --from-beginning --   注意：此参数有生效条件；1.换一个消费组2.offset在服务器中不存在（过期删除）
        return props;
    }
}
